Operating on a dynamic 3-minute timeframe, Adaptive Edge pioneers precision in short-term trading. With a keen eye on market anomalies, it not only navigates fluctuations strategically but also boasts remarkable gains, showcasing a winning formula for high-performance trading.
We begin by importing the necessary packages for data processing and visualizing the results and plots.
import pandas as pd
import pandas_ta as ta
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from datetime import datetime
The provided dataset is loaded. If using this code version, ensure to modify the csv_path to the correct file path after downloading the dataset. The dataset is described as a 1-minute OHLC dataset, including information such as Date, Expiry Date of the contract, and Time. Appropriate header files are added for clarity as shown below.
csv_path = "/Users/siddharthacharya/Downloads/Data_2020-2022_wExpiry.csv"
columns = ['Date', 'ExpiryDate', 'Time', 'Open', 'High', 'Low', 'Close']
df_1 = pd.read_csv(csv_path, header=None, names=columns)
df_1 = df_1.dropna()
df_1.tail()
| Date | ExpiryDate | Time | Open | High | Low | Close | |
|---|---|---|---|---|---|---|---|
| 277870 | 20221230 | 20230125 | 1526 | 23110.54002 | 23144.45146 | 23108.56403 | 23135.66988 |
| 277871 | 20221230 | 20230125 | 1527 | 23133.20795 | 23134.09097 | 23110.70025 | 23111.47473 |
| 277872 | 20221230 | 20230125 | 1528 | 23111.82193 | 23121.38728 | 23104.82660 | 23121.38728 |
| 277873 | 20221230 | 20230125 | 1529 | 23118.95512 | 23134.30505 | 23111.82193 | 23122.91099 |
| 277874 | 20221230 | 20230125 | 1530 | 23122.45653 | 23132.08420 | 23120.63887 | 23130.66629 |
Our trading strategy relies on 3 minute closing prices to make decisions and find the winnning patterns. To make our model work, we first have to change the given 1-minute data into the 3 minute timeframe format. This was already discussed in the previous strategy. Here's how we do it:
# Preserve the original dataset
dfc = df_1.copy()
# Convert 'Date' column to a datetime format
dfc['Date'] = pd.to_datetime(dfc['Date'], format='%Y%m%d')
# Extract hour and minute information from the 'Time' column
dfc['Hour'] = dfc['Time'] // 100
dfc['Minute'] = dfc['Time'] % 100
# Create a new datetime column combining date, hour, and minute
dfc['DateTime'] = pd.to_datetime(dfc[['Date', 'Hour', 'Minute']].astype(str).agg('-'.join, axis=1), format='%Y-%m-%d-%H-%M')
# Set 'DateTime' column as the index
dfc.set_index('DateTime', inplace=True)
def df_tf(t):
# Resample the data to desired intervals and use the open price as the resampling method
df_tf = dfc.resample(f'{t}',closed = 'right').agg({'Open': 'first', 'High': 'max', 'Low': 'min', 'Close': 'last'})
df_tf.dropna(subset=['Close'], inplace=True)
# Reset the index to make 'DateTime' a regular column again
df_tf.reset_index(inplace=True)
return df_tf
In this section, the focus is on calculating the change in the closing price from two candles ago. This is achieved by creating a new column:2_Candle_Signal. The 2_Candle_Signal column, captures whether the difference in closing prices from two candles ago is greater than 10 points.
The choice of 10 points as a threshold is strategic. In this context, it is essential to consider the trading scenario. With a quantity of 100, the minimum price difference required to cover a commission of 750 is 7.5 points. To simplify, a threshold of 10 points is chosen, creating three distinct bins: below -10, above 10, and in between. The 2_Candle_Signal thus provides a signal for each candle, indicating whether the price difference from two candles ago was greater than 10 points, less than -10 points, or within the -10 to 10 point range. This strategic binning facilitates decision-making, helping to identify significant price movements and potential profit opportunities.
df_3T = df_tf('3T')
df = df_3T.copy()
# Trades taken at closing price
df_3T['DateTime'] = df['DateTime'] + pd.Timedelta(minutes=3)
df_3T['1_Candle_Change'] = df_3T['Close'] - df_3T['Close'].shift(1)
df_3T['2_Candle_Change'] = df_3T['Close'] - df_3T['Close'].shift(2)
# Classify the change based on the threshold
df_3T['1_Candle_Signal'] = pd.cut(df_3T['1_Candle_Change'], bins=[-float('inf'), -10, 10, float('inf')],
labels=['Down', 'None', 'Up'])
df_3T['2_Candle_Signal'] = pd.cut(df_3T['2_Candle_Change'], bins=[-float('inf'), -10, 10, float('inf')],
labels=['Down', 'None', 'Up'])
df_3T = df_3T.dropna()
Almost all scalping stragies thrive on momentum. In our case we use the Moving Average Convergence Divergence (MACD) technical indicator to filter the bullish and bearish signals in the market.
# Calculate the MACD
def calculate_macd(df, fast_period=12, slow_period=26, signal_period=9):
df['fast_ema'] = df['Close'].ewm(span=fast_period, min_periods=1, adjust=False).mean()
df['slow_ema'] = df['Close'].ewm(span=slow_period, min_periods=1, adjust=False).mean()
df['macd'] = df['fast_ema'] - df['slow_ema']
df['signal_line'] = df['macd'].ewm(span=signal_period, min_periods=1, adjust=False).mean()
df['macd_histogram'] = df['macd'] - df['signal_line']
calculate_macd(df_3T)
# Generate Buy and Sell signals based on your conditions
df_3T['buy_signal'] = (df_3T['macd'] > 0) & (df_3T['macd'].shift(1) <= 0) & (df_3T['macd'] > df_3T['signal_line'])
df_3T['sell_signal'] = (df_3T['macd'] < 0) & (df_3T['macd'].shift(1) >= 0) & (df_3T['macd'] < df_3T['signal_line'])
class Backtester:
def __init__(self):
# Initialize strategy parameters and variables
self.long_position = {'active': False, 'entry_price': 0.0, 'exit_price': 0.0, 'trade_pnl': 0.0}
self.short_position = {'active': False, 'entry_price': 0.0, 'exit_price': 0.0, 'trade_pnl': 0.0}
self.current_pnl = 0.0 # Track current pnl for the active position
self.pnl = 0.0
self.total_pnl = []
self.trades = []
def execute_trade(self, trade_type, entry_price, entry_time):
# Execute a trade and update relevant variables
trade = {'type': trade_type, 'entry_price': entry_price, 'entry_time': entry_time,
'exit_price': None, 'exit_time': None, 'trade_pnl': 0.0}
self.trades.append(trade)
if trade_type == 'long':
self.long_position['active'] = True
self.long_position['entry_price'] = entry_price
elif trade_type == 'short':
self.short_position['active'] = True
self.short_position['entry_price'] = entry_price
else:
trade_pnl = 0
self.total_pnl.append(self.pnl)
def close_position(self, exit_price,exit_time):
# Close the active position and update relevant variables
if self.long_position['active']:
self.long_position['active'] = False
self.long_position['exit_price'] = exit_price
self.long_position['trade_pnl'] = 100 * (exit_price - self.long_position['entry_price']) - 750
self.pnl += self.long_position['trade_pnl']
self.current_pnl = 0.0 # Reset current pnl after exiting
# Update exit price and pnl in the trades list
self.trades[-1]['exit_price'] = exit_price
self.trades[-1]['exit_time'] = exit_time
self.trades[-1]['trade_pnl'] = self.long_position['trade_pnl']
self.total_pnl.append(self.pnl)
elif self.short_position['active']:
self.short_position['active'] = False
self.short_position['exit_price'] = exit_price
self.short_position['trade_pnl'] = -100 * (exit_price - self.short_position['entry_price']) - 750
self.pnl += self.short_position['trade_pnl']
self.current_pnl = 0.0 # Reset current pnl after exiting
# Update exit price and pnl in the trades list
self.trades[-1]['exit_price'] = exit_price
self.trades[-1]['exit_time'] = exit_time
self.trades[-1]['trade_pnl'] = self.short_position['trade_pnl']
self.total_pnl.append(self.pnl)
def update_current_pnl(self, current_price):
# Update the current pnl for the active position
if self.long_position['active']:
self.current_pnl = 100 * (current_price - self.long_position['entry_price']) - 750
elif self.short_position['active']:
self.current_pnl = -100 * (current_price - self.short_position['entry_price']) - 750
else:
self.current_pnl = 0
def get_summary(self):
# Return a summary of the strategy's performance
summary = {
'total_pnl': self.total_pnl,
'trades': self.trades,
'pnl': self.pnl,
'current_pnl': self.current_pnl
}
return summary
The inclusion of an example transition matrix in this report is aimed at providing a clear demonstration of the pervasive presence of market edges. Post-strategy development, the matrix serves as a tool for retrospective analysis without introducing any look-ahead bias. The examples within underscore the existence of anomalies, from straightforward price differences to nuanced patterns. This nuanced understanding of market dynamics reinforces the adaptability and effectiveness of our scalping strategy in identifying and leveraging these ever-changing edges for optimal trading outcomes.
transition_matrix = pd.crosstab(df_3T['2_Candle_Signal'].shift(), df_3T['2_Candle_Signal'], normalize='index')
transition_matrix
| 2_Candle_Signal | Down | None | Up |
|---|---|---|---|
| 2_Candle_Signal | |||
| Down | 0.567426 | 0.218683 | 0.213892 |
| None | 0.305092 | 0.387504 | 0.307403 |
| Up | 0.220988 | 0.215333 | 0.563678 |
We need to create a list of the expiry dates in order to exit them on time.
import pandas as pd
df_1['ExpiryDate'] = pd.to_datetime(df_1['ExpiryDate'], format='%Y%m%d', errors='coerce')
unique_expiry_dates = df_1['ExpiryDate'].unique().tolist()
unique_expiry_dates=pd.to_datetime(unique_expiry_dates)
unique_expiry_dates
DatetimeIndex(['2020-01-30', '2020-02-27', '2020-03-26', '2020-04-30',
'2020-05-28', '2020-06-25', '2020-07-30', '2020-08-27',
'2020-09-24', '2020-10-29', '2020-11-26', '2020-12-31',
'2021-01-28', '2021-02-25', '2021-03-25', '2021-04-29',
'2021-05-27', '2021-06-24', '2021-07-29', '2021-08-26',
'2021-09-30', '2021-10-28', '2021-11-25', '2021-12-30',
'2022-01-27', '2022-02-24', '2022-03-31', '2022-04-28',
'2022-05-26', '2022-06-30', '2022-07-28', '2022-08-25',
'2022-09-29', '2022-10-27', '2022-11-24', '2022-12-29',
'2023-01-25'],
dtype='datetime64[ns]', freq=None)
Comments in the code provide clarity by explaining each part of the logic, making it easy to understand and work with.
# Initialize Backtester
backtester = Backtester()
# Lists to store capital and count of consecutive Up and Down signals
capital = []
u, d = 0, 0
# Define time intervals for scalping
time1 = pd.to_datetime('9:21').time()
time2 = pd.to_datetime('10:00').time()
time3 = pd.to_datetime('11:00').time()
time4 = pd.to_datetime('11:00').time()
time5 = pd.to_datetime('14:30').time()
exit_time = pd.to_datetime('15:00').time()
# Copy the original DataFrame for processing
df = df_3T.copy()
# Extract time and date information from the DateTime column
df['DateTime'] = pd.to_datetime(df['DateTime'])
df['Time'] = df['DateTime'].dt.time
df['Date'] = df['DateTime'].dt.date
# Loop through the DataFrame and execute trading strategy
for i, data in df.iterrows():
# Update current PnL
backtester.update_current_pnl(data['Close'])
# Check trading time windows
if data['Time'] <= time1 or time2 < data['Time'] < time3 or time4 < data['Time'] <= time5:
# Execute Short trade on Up signal
if data['2_Candle_Signal'] == 'Up' and not data['buy_signal']:
u += 1
# Exit if conditions met or in an active long position
if (d > 4 or backtester.current_pnl < -15000) and backtester.long_position['active']:
backtester.close_position(data['Close'], data['DateTime'])
# Execute Short trade if conditions met
if u < 3 and not (backtester.long_position['active'] or backtester.short_position['active']):
backtester.execute_trade('short', data['Close'], data['DateTime'])
d = 0
# Execute Long trade on Down signal
elif data['2_Candle_Signal'] == 'Down' and not data['sell_signal']:
d += 1
# Exit if conditions met or in an active short position
if (u > 4 or backtester.current_pnl < -15000) and backtester.short_position['active']:
backtester.close_position(data['Close'], data['DateTime'])
# Execute Long trade if conditions met
if d < 3 and not (backtester.long_position['active'] or backtester.short_position['active']):
backtester.execute_trade('long', data['Close'], data['DateTime'])
u = 0
# Check exit conditions based on time
if data['Time'] >= exit_time:
backtester.close_position(data['Close'], data['DateTime'])
# Record capital at each step
capital.append(backtester.pnl + backtester.current_pnl)
# Get the final PnL and print
summary = backtester.get_summary()
final_pnl = summary['pnl']
print(f"Final PnL: {final_pnl}")
Final PnL: 8693804.322000023
print(f"Profits without commission: {750*len(summary['trades'])+final_pnl}")
Profits without commission: 10285304.322000023
import plotly.graph_objects as go
# Create a Plotly figure
fig = go.Figure()
# Add a line trace for the different performances
fig.add_trace(go.Scatter(x=df['DateTime'], y=capital, mode='lines', name='Algo Performance',line=dict(color='blue')))
fig.add_trace(go.Scatter(x=df['DateTime'], y=(df['Close'][2:]-df['Close'][2])*100, mode='lines', name='Buy and Hold',line=dict(color='green')))
# Customize the layout
fig.update_layout(
title='Algo vs. Market: PnL Showdown',
xaxis_title='Week #',
yaxis_title='Cumulative PnL',
template='seaborn',
)
# Display plot
fig.show()